package drds.data_migrate.job;


import drds.common.Author;
import drds.data_migrate.Authors;
import drds.data_migrate.task_group.TaskGroupContainer;
import drds.data_migrate.task_group.TaskGroupTask;
import drds.data_migrate.vo.State;
import drds.data_migrate.vo.Status;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Validate;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Author(name = Authors.LI_YANG)
@Slf4j
public class Scheduler
{


    @Setter
    @Getter
    private Long jobId;
    @Setter
    @Getter
    private ExecutorService executorService;


    public void schedule(List<TaskGroup> configList)
    {
        Validate.notNull(configList, "scheduler配置不能为空");
        this.jobId = configList.get(0).getJobId();
        int jobSleepIntervalInMillSec = 0;//configList.get(0).getJobSleepInterval();// .getInt(Constants.jobSleepInterval, 10000);
        int jobReportIntervalInMillSec = 0;//configList.get(0).getJobReportInterval();//.getInt(Constants.jobReportInterval, 30000);


        int totalTasks = calculateTaskCount(configList);
        startAllTaskGroup(configList);

        State state = new State();

        long lastReportTimeStamp = System.currentTimeMillis();
        try
        {
            while (true)
            {
                /**
                 * step 1: collect job stat
                 * step 2: getReport info, then report it
                 * step 3: errorRecordChecker do check
                 * step 4: dealSucceedStat();
                 * step 5: dealKillingStat();
                 * step 6: dealFailedStat();
                 * step 7: refresh last job stat, and then sleep for next while
                 *
                 * above steps, some ones should report info to DS
                 *
                 */


                //汇报周期
                long now = System.currentTimeMillis();
                if (now - lastReportTimeStamp > jobReportIntervalInMillSec)
                {

                }


                if (state.getStatus() == Status.succeeded)
                {
                    log.info("Scheduler accomplished all wrapper.");
                    break;
                }

                if (isJobKilling(this.getJobId()))
                {
                    dealKillingStat();
                } else if (state.getStatus() == Status.failed)
                {
                    dealFailedStat(state.getThrowable());
                }

                Thread.sleep(jobSleepIntervalInMillSec);
            }
        } catch (InterruptedException e)
        {
            // 以 failed 状态退出
            log.error("捕获到InterruptedException异常!", e);

            throw new RuntimeException(e);
        }

    }

    protected void startAllTaskGroup(List<TaskGroup> configList)
    {
        this.executorService = Executors.newFixedThreadPool(configList.size());

        for (TaskGroup config : configList)
        {
            TaskGroupTask taskGroupTask = newTaskGroupTask(config);
            this.executorService.execute(taskGroupTask);
        }

        this.executorService.shutdown();
    }

    private TaskGroupTask newTaskGroupTask(
            TaskGroup config)
    {
        TaskGroupContainer taskGroupContainer = new TaskGroupContainer(config);

        return new TaskGroupTask(taskGroupContainer);
    }

    protected void dealFailedStat(Throwable throwable)
    {
        this.executorService.shutdownNow();
        throw new RuntimeException();
    }

    protected void dealKillingStat()
    {
        //通过进程退出返回码标示状态
        this.executorService.shutdownNow();
        throw new RuntimeException("job killed status");
    }

    private int calculateTaskCount(List<TaskGroup> configList)
    {
        int totalTasks = 0;
        for (TaskGroup config : configList)
        {
            totalTasks += config.getTaskConfigList().size();
        }
        return totalTasks;
    }


    protected boolean isJobKilling(Long jobId)
    {
        return false;
    }
}
